package ws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import netty.util.StartAndStop;
import okhttp3.*;
import okio.ByteString;
import test.PerformanceAnalyze;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Proxy;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class OkHttpBinanceWsPublic extends StartAndStop {

    static PerformanceAnalyze performanceAnalyze = new PerformanceAnalyze();

    private WebSocket init(String url) {
        OkHttpClient mClient = new OkHttpClient.Builder()
                .readTimeout(30, TimeUnit.SECONDS)//设置读取超时时间
                .writeTimeout(30, TimeUnit.SECONDS)//设置写的超时时间
                .connectTimeout(30, TimeUnit.SECONDS)//设置连接超时时间
                .proxy(new Proxy(Proxy.Type.HTTP, new InetSocketAddress("127.0.0.1", 1087)))
                .build();

        //构建一个连接请求对象
        Request request = new Request.Builder().get().url(url).build();

        //开始连接
        return mClient.newWebSocket(request, new WebSocketListener() {
            @Override
            public void onOpen(WebSocket webSocket, Response response) {
                super.onOpen(webSocket, response);
                System.out.println("连接成功, webSocket: " + webSocket);
                System.out.println("连接成功, response: " + webSocket);
            }

            @Override
            public void onMessage(WebSocket webSocket, String text) {
                super.onMessage(webSocket, text);
                performanceAnalyze.meter();
                performanceAnalyze.length(text.length());

                System.out.println(text);

                JSONObject jsonObject = JSON.parseObject(text);
                JSONObject data = jsonObject.getJSONObject("data");
                if (data != null) {
                    long e = jsonObject.getJSONObject("data").getLongValue("E");
                    if (e != 0) {
                        performanceAnalyze.delay(System.currentTimeMillis() - e);
                    }
                }
            }

            @Override
            public void onMessage(WebSocket webSocket, ByteString bytes) {
                super.onMessage(webSocket, bytes);
                System.out.println("收到消息, webSocket: " + webSocket);
                System.out.println("收到消息, bytes: " + bytes);
            }

            @Override
            public void onClosed(WebSocket webSocket, int code, String reason) {
                super.onClosed(webSocket, code, reason);
                System.out.println("连接关闭..., webSocket: " + webSocket);
                System.out.println("连接关闭..., code" + code);
                System.out.println("连接关闭..., reason" + reason);
            }

            @Override
            public void onFailure(WebSocket webSocket, Throwable throwable, Response response) {
                super.onFailure(webSocket, throwable, response);
                System.out.println("连接失败..., webSocket: " + webSocket);
                throwable.printStackTrace();
                System.out.println("连接失败..., response: " + response);
            }
        });
    }

    public void start(String url, int size) throws IOException {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);

        List<WebSocket> wsList = new LinkedList<>();

        OkHttpBinanceWsPublic okHttpWsPublic = new OkHttpBinanceWsPublic();

        for (int i = 0; i < size; i++) {
            wsList.add(okHttpWsPublic.init(url));
        }

        for (WebSocket webSocket : wsList) {
            executor.scheduleAtFixedRate(() -> webSocket.send("ping"), 1, 10, TimeUnit.SECONDS);
        }
    }

}
